package priv.tuyou.myself;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class StreamApiTest {

	public static void main(String[] args) throws Exception {
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStream<Long> someIntegers = env.generateSequence(0, 1000);

		IterativeStream<Long> iteration = someIntegers.iterate();

		DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
		  @Override
		  public Long map(Long value) throws Exception {
//			  System.out.println("111:" + value);
//			  Thread.sleep(1000);
		    return value - 1 ;
		  }
		});

		DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
		  @Override
		  public boolean filter(Long value) throws Exception {
//			  System.out.println("22222222:" + value);
//			  Thread.sleep(1000);
		    return (value > 0);
		  }
		});

		iteration.closeWith(stillGreaterThanZero);

		DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
		  @Override
		  public boolean filter(Long value) throws Exception {
			  System.out.println("33333333333333:" + value);
//			  Thread.sleep(1000);
		    return (value <= 0);
		  }
		});
		
		env.execute();
		
	}
}
